package online

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

case class ProductRating(userId: Int, productId: Int, rating: Double, timestamp: Long)

case class MongoConfig(uri: String, db: String)

// 定义标准推荐对象
case class Recommendation(productId: Int, score: Double)

// 定义商品相似度列表
case class ProductRecs(productId: Int, recs: Seq[Recommendation])


// 定义一个连接助手对象，建立到redis和mongodb的连接
object ConnHelper extends Serializable {
  // 懒变量定义，使用的时候才初始化
  lazy val jedis = new Jedis("101.33.204.176")
  jedis.auth("lxj2002617")
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://big_data:Bigdata%402023@dds-bp1a4fbded1f3ee41316-pub.mongodb.rds.aliyuncs.com:3717/big_data"))
}

object OnLineRecommender {
  // 定义常量和表名
  val MONGODB_RATING_COLLECTION = "Rating" //需要读取商品的评分表
  val STREAM_RECS = "StreamRecs" //实时推荐结果将存储到该集合中
  val PRODUCT_RECS = "ProductRecs" //实时推荐基于该商品相似矩阵

  val MAX_USER_RATING_NUM = 20 //评分数最高的20个商品
  val MAX_SIM_PRODUCTS_NUM = 20 //相似度最高的20个商品

  //主函数
  def main(args: Array[String]): Unit = {
    try {
      //MongoDB和Kafka的连接主题配置
      val config = Map(
        "mongo.uri" -> "mongodb://big_data:Bigdata%402023@dds-bp1a4fbded1f3ee41316-pub.mongodb.rds.aliyuncs.com:3717/big_data",
        "mongo.db" -> "big_data",
        "kafka.topic" -> "recommender" //kafka中实时打分数据流主题
      )

      // 创建spark conf
      val sparkConf = new SparkConf().setMaster("local[*]").setAppName("OnlineRecommender")
      val spark = SparkSession.builder().config(sparkConf).getOrCreate()
      val sc = spark.sparkContext
      val ssc = new StreamingContext(sc, Seconds(2))
      //引入隐士
      import spark.implicits._
      implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

      //读取Mongo中离线推荐准备的商品相似矩阵
      val simProductsMatrix = spark
        .read
        .option("uri", mongoConfig.uri)
        .option("collection", PRODUCT_RECS)
        .format("com.mongodb.spark.sql")
        .load()
        .as[ProductRecs]
        .rdd
        // 为了后续查询相似度方便，把数据转换成map形式
        .map { item =>
          (item.productId, item
            .recs //取商品的相似商品列表
            .map(x => (x.productId, x.score)) //(商品Id,商品和商品的相似度)
            .toMap //将(商品Id,商品和商品的相似度)转换成Map
          )
        }
        .collectAsMap() //将上诉数据搜集为Map格式
      //simProductsMatrix.foreach(println(_))
      //将simProductsMatrix广播出去
      val simProcutsMatrixBC = sc.broadcast(simProductsMatrix)

      // 创建kafka配置参数
      val kafkaParam = Map(
        "bootstrap.servers" -> "192.168.10.101:9092", //kafka集群地址
        "key.deserializer" -> classOf[StringDeserializer], //从kafka中读取key数据反序列化方式
        "value.deserializer" -> classOf[StringDeserializer], //从kafka中读取value数据反序列化方式
        "group.id" -> "recommender", //消费者组ID
        "auto.offset.reset" -> "latest" //从主题的最近offset开始消费
      )

      // 创建一个DStream
      val kafkaStream = KafkaUtils.createDirectStream[String, String](
        ssc, //StreamingContext上下文对象
        LocationStrategies.PreferConsistent, //本地策略
        ConsumerStrategies.Subscribe[String, String](Array(config("kafka.topic")), kafkaParam) //消费策略
      )

      // 对kafka中实时打分数据进行处理，产生评分流，userId|productId|score|timestamp
      val ratingStream = kafkaStream.map { msg =>
        val attr = msg.value() //.value()获取Kafka的数据 如：3061645|14103|5.0|1655954396
          .split("\\|") //对数据流进行拆分
        (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt) //数据封装
      }
      //ratingStream.print()
      // 核心算法部分，定义评分流的处理流程
      ratingStream.foreachRDD {
        rdds =>
          rdds.foreach {
            case (userId, productId, score, timestamp) =>
              println("rating data coming!>>>>>>>>>>>>>>>>>>")

              // TODO: 核心算法流程
              // 1. 从redis里取出当前用户的最近评分，保存成一个数组Array[(productId, score)]
              val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATING_NUM, userId, ConnHelper.jedis)
              //userRecentlyRatings.foreach(println(_))

              // 2. 从商品的相似度矩阵中获取当前商品最相似的商品列表，作为备选列表，保存成一个数组Array[productId]
              val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId, userId, simProcutsMatrixBC.value)
              //candidateProducts.foreach(println(_))

              // 3. 计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表，保存成 Array[(productId, score)]
              val streamRecs = computeProductScore(candidateProducts, userRecentlyRatings, simProcutsMatrixBC.value)
              //streamRecs.foreach(println(_))

              // 4. 把推荐列表保存到mongodb
              saveDataToMongoDB(userId, streamRecs)
          }
      }
      //ssc开启
      ssc.start()
      //等待结束
      ssc.awaitTermination()
    }
  }

  /**
   * 从redis里获取最近num次评分
   */

  import scala.collection.JavaConversions._ //java代码和scala代码互转包

  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {
    // 从redis中用户的评分队列里获取评分数据，list键名为uid:USERID，值格式是 PRODUCTID:SCORE
    jedis.lrange("userId:" + userId.toString, 0, num) //因为这儿返回的是Java版本的List，所以需要进行Java代码转换scala
      .map { item =>
        val attr = item.split("\\:") //使用:拆分成商品ID和评分
        (attr(0).trim.toInt, attr(1).trim.toDouble) //封装(productId,score)
      }
      .toArray //将(productId,score)封装成数组 [(productId,score),(productId,score),(productId,score)]
  }


  // 获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表
  def getTopSimProducts(num: Int, //20个最相似
                        productId: Int,
                        userId: Int,
                        simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                       (implicit mongoConfig: MongoConfig): Array[Int] = {
    // 从广播变量相似度矩阵中拿到当前商品的相似度列表
    val allSimProducts = simProducts(productId).toArray //将打分的商品ID的相似商品取出，并转换为数组

    // 获取mongodb中Rating集合的连接客户端
    val ratingCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    //从Rating集合中获取当前登陆用户评过分的productId
    val ratingExist = ratingCollection
      .find(MongoDBObject("userId" -> userId)) //从集合中查询db.Rating.find({"userId"->userId})
      .toArray //转换成数组
      .map { item =>
        item.get("productId").toString.toInt //抽选出productId
      }

    // 从商品相似矩阵中过滤掉已经评分过的商品--->并且按照相似度进行到排序--->最后取出相似度最高的前20个商品返回
    allSimProducts
      .filter(x => !ratingExist.contains(x._1)) //从商品相似矩阵中过滤掉已经评分过的商品，避免重复推荐
      .sortWith(_._2 > _._2) //从商品相似矩阵中过滤掉已经评分过的商品
      .take(num) //最后取出相似度最高的前20个商品返回
      .map(x => x._1) //封装商品Id返回
  }

  // 自定义log函数，以N为底
  def log(m: Int): Double = {
    val N = 10
    math.log(m) / math.log(N)
  }

  //获取产品相似度
  def getProductsSimScore(product1: Int,
                          product2: Int,
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    simProducts.get(product1) match { //根据product1取出对应的(商品->相似度)
      case Some(sims) => sims.get(product2) match { //在根据product2取出对应的相似度
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }


  // 计算每个备选商品的推荐得分
  def computeProductScore(candidateProducts: Array[Int],
                          userRecentlyRatings: Array[(Int, Double)],
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {
    // 定义一个长度可变数组ArrayBuffer，用于保存每一个备选商品的基础得分，(productId, score)
    val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    // 定义两个map，用于保存每个商品的高分和低分的计数器，productId -> count
    val increMap = scala.collection.mutable.HashMap[Int, Int]() //评分大于3分存储集合
    val decreMap = scala.collection.mutable.HashMap[Int, Int]() //评分小于3分的存储集合

    // 遍历每个备选商品，计算和已评分商品的相似度
    for (candidateProduct <- candidateProducts; userRecentlyRating <- userRecentlyRatings) {
      // 从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度值
      val simScore = getProductsSimScore(candidateProduct, userRecentlyRating._1, simProducts)
      if (simScore > 0.4) { //排除相似度值小于0.4的商品
        // 按照公式进行加权计算，得到基础评分，然后将(候选商品,相似度与打分乘积)添加到scores集合中
        scores += ((candidateProduct, simScore * userRecentlyRating._2))
        //根据对备选商品打分次数值进行衰减
        if (userRecentlyRating._2 > 3) {
          increMap(candidateProduct) = increMap.getOrDefault(candidateProduct, 0) + 1
        } else {
          decreMap(candidateProduct) = decreMap.getOrDefault(candidateProduct, 0) + 1
        }
      }
    }

    // 根据公式计算所有的推荐优先级，首先以productId做groupby
    scores
      .groupBy(_._1) //根据备选商品ID进行分组
      .map {
        case (productId, scoreList) =>
          //封装数据
          (productId,
            scoreList.map(_._2).sum / scoreList.length //平均值
              + log(increMap.getOrDefault(productId, 1)) //加强分
              - log(decreMap.getOrDefault(productId, 1)) //削弱分
          )
      }
      .toArray //转换成数组
      .sortWith(_._2 > _._2) //根据备选商品最后的优先级进行倒排序
  }

  // 写入mongodb
  def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit = {
    //获取StreamRecs集合连接客户端
    val streamRecsCollection = ConnHelper.mongoClient(mongoConfig.db)(STREAM_RECS)
    // 按照userId查询并更新
    if (streamRecs.size > 0) {
      streamRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
    }
    //新增(userId,(productId->推荐优先级))
    streamRecsCollection.insert(MongoDBObject("userId" -> userId,
      "recs" -> streamRecs.map(x => MongoDBObject("productId" -> x._1, "score" -> x._2))))
  }
}
